package com.duowan.realtime.thrift;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.duowan.realtime.thirft.api.Constants;
import com.duowan.realtime.thirft.api.HyperLogLogException;
import com.duowan.realtime.thirft.api.HyperLogLogPlusException;
import com.duowan.realtime.thirft.api.HyperLogLogQuery;
import com.duowan.realtime.thirft.api.HyperLogLogService.Iface;

public class HyperLogLogServiceImpl implements Iface {

	private static final Logger LOG = LoggerFactory.getLogger("hlllog");
	
	private ProxyHyperLogLogServiceImpl proxyHllImpl;
	public HyperLogLogServiceImpl(){
		proxyHllImpl = new ProxyHyperLogLogServiceImpl();
	}

	@Override
	public String ping() throws HyperLogLogPlusException, TException {
		return Constants.PING_RESPONSE;
	}

	@Override
	public Map<String, Integer> offerForCardinality(String hllGroup, List<HyperLogLogQuery> hllQueryList) throws HyperLogLogException, TException {
		Map<String, Integer> resultMap = new ConcurrentHashMap<String, Integer>();
		for (HyperLogLogQuery hllQuery : hllQueryList) {
			int count = (int)proxyHllImpl.offerForCardinality(hllGroup, hllQuery.getGroup(), hllQuery.getValues());
			resultMap.put(hllQuery.getGroup(), count);
			LOG.info("HllpQuery values size:" + hllQuery.getValues().size() + ",result:" + hllQuery.getGroup() + " and current cardinality=" + count + " into Hyperloglog pool");
		}
		return resultMap;
	}

	@Override
	public void offer(String hllGroup, List<HyperLogLogQuery> hllQueryList) throws HyperLogLogException, TException {
		for (HyperLogLogQuery hllQuery : hllQueryList) {
			proxyHllImpl.offer(hllGroup, hllQuery.getGroup(), hllQuery.getValues());;
			LOG.info("HllpQuery group " + hllQuery.getGroup() + ",values size:" + hllQuery.getValues().size());
		}
	}

	@Override
	public long cardinality(String hllGroup, String group) throws HyperLogLogException, TException {
		long estimate = proxyHllImpl.cardinality(hllGroup, group);
		LOG.info("hllpGroup=" + hllGroup + ", query group=" + group + ",cardinality estimate=" + estimate);
		return estimate;
	}

	@Override
	public List<Map<String, String>> keys(String hllGroup, String key)throws HyperLogLogException, TException {
		LOG.info("get keys:" + "hllGroup=" + hllGroup + ",key=" + key);
		return proxyHllImpl.keys(hllGroup, key);
	}




}
